package com.maxt.common.mqtt;

import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.maxt.controller.DemoController;

public class PublishClient {
	
	public static final Logger logger = LoggerFactory.getLogger(PublishClient.class);
	
    public static final String HOST = "tcp://ext.prod.emqtt1.iwangfan.cn:1883";
    private static final String clientid =  "coolId" ;// Heartbeat.currentVersion + Heartbeat.D.getID();
    private static MqttAsyncClient client;

    private static final String USERNAME = "dma@emqtt" ;   //"CofEIo/9YPhu" ;    // "dma@emqtt"
    private static final String PASSWORD = "gfyybkj1" ;  //"C2Cx0mA15GT=" ;    // "gfyybkj1" 
    
    public static PublishClient create(){
        PublishClient client = new PublishClient();     
        return client;
    }
    private PublishClient() {
        // MemoryPersistence设置clientid的保存形式，默认为以内存保存
        try { 
            client = new MqttAsyncClient(HOST, clientid, new MemoryPersistence()); 
            doconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private void doconnect() {
        // Tokens
        IMqttToken connectToken;
        
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(USERNAME);  // new String(HMBase64.decode( USERNAME )));
        options.setPassword(PASSWORD.toCharArray()); //new String(HMBase64.decode( PASSWORD )).toCharArray());
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        // 设置超时时间
        options.setConnectionTimeout(20);
        // 设置会话心跳时间
        options.setKeepAliveInterval(10);
        
        DisconnectedBufferOptions disconnectedOpts = new DisconnectedBufferOptions();
        disconnectedOpts.setBufferEnabled(true);
        disconnectedOpts.setBufferSize(1000); // 1000 message buffer
        disconnectedOpts.setDeleteOldestMessages(true); // Purge oldest messages when buffer is full
        disconnectedOpts.setPersistBuffer(false); // Do not buffer to disk
        client.setBufferOpts(disconnectedOpts);

        // Enable Proxy & Connect to server 
        try {
            connectToken = client.connect(options);
            connectToken.waitForCompletion();
            //boolean isConnected = client.isConnected();

        } catch (Exception e) {
            logger.info("[mqtt] client.connect failed!" + e.getMessage());
        }

        client.setCallback(new MqttCallback() {
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
            	logger.info("[mqtt] messageArrived ",topic);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
            	logger.info("[mqtt] deliveryComplete");
            }

            @Override
            public void connectionLost(Throwable cause) {
            	logger.info("[mqtt] reconnected");
            }
        });
    }

    public void close() {
        try {
            // 等到有未完成的工作完成是关闭连接
        	logger.info("[mqtt] client disconnect");
            client.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    // 发布空消息
    public static void publish(String topic )  {
        publish(  topic,new JsonObject());
    }
    
    // 发布
    public static void publish(String topic,String jsonStr )  {
        JsonObject message = new JsonObject();
        try {
            message = new JsonParser().parse(jsonStr).getAsJsonObject();
        } catch (Exception e) {
        	logger.info(e.getMessage());
        }
        publish( topic, message );
    }
    
    // String topic1 = "dma/flow/{deviceSN}/onwifi";
    // String topic2 = "dma/flow/{deviceSN}/offwifi";
    // String topic3 = "dma/flow/{deviceSN}/onInternet";
    // String topic4 = "dma/flow/{deviceSN}/offInternet";
    // String topic5 = "dma/flow/{deviceSN}/traffic";
    public static int messageCount = 0;
    public static void publish(String topic, JsonObject message  )  {
    	logger.info("[mqtt] sending "+topic);
        try {
            JsonObject obj = new JsonObject();
            obj.addProperty("time", System.currentTimeMillis());
            obj.addProperty("dev", "GD201AAAAFFFF") ;  //;Heartbeat.D.getID());
            obj.addProperty("topic", topic);
            obj.addProperty("count", messageCount++);
            obj.add("msg", message);
            MqttMessage msg = new MqttMessage(new Gson().toJson(obj).getBytes()) ;
            msg.setQos(1);
            if(client != null){
                client.publish(topic, msg);
                logger.info("[mqtt] publishing "+topic);
            }  else {
            	logger.info("[mqtt] can not publishing "+topic);
            }
        } catch (MqttPersistenceException e) { 
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        } catch (UnsupportedOperationException e) {
        	logger.info("[mqtt] can not publishing "+topic);
            e.printStackTrace();
            
        }
    }
    
    public static void  main(String[] args){
    	create();
    	publish("cool_999","{\"name\":\"zhangsan\",\"age\":\"21\"}");
    }

}